package com.zheng.cms.rocketmq.consumer;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.TransactionCheckListener;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import org.apache.log4j.Logger;

import java.util.List;

public class TranscationalConsumer {


    private Logger log = Logger.getLogger(getClass());

    private DefaultMQPushConsumer consumer;

    private String consumerGroup;

    private String namesrvAddr;

    private String instanceName;

    private String topic;




    /**
     * 设置一个Listener，主要进行消息的逻辑处理
     *注意这里使用的是MessageListenerOrderly这个接口
     * */
    public TranscationalConsumer(String consumerGroup, String namesrvAddr, String instanceName, String topic) {
        this.consumerGroup = consumerGroup;
        this.namesrvAddr = namesrvAddr;
        this.instanceName = instanceName;
        this.topic = topic;

    }
    /**
     * //这里设置的是一个consumer的消费策略
     *         //CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
     *         //CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
     *         //CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
     *         consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
     *
     *         设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
     *          * 如果非第一次启动，那么按照上次消费的位置继续消费
     *   consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
     * */
    public void init() throws MQClientException {
        log.info("start init DefaultMQPushConsumer...");
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); //从队列头部开始消费
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setInstanceName(instanceName);
        consumer.subscribe(topic, "*");//订阅所有
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                try {

                    for (MessageExt msg : msgs) {
                        System.out.println(msg + ",内容：" + new String(msg.getBody()));
                    }

                } catch (Exception e) {
                    e.printStackTrace();

                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试

                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功

                }
            });

        consumer.start();
        log.info("DefaultMQPushConsumer init ok.");
    }

    public void destroy(){
        log.info("start destroy DefaultMQPushConsumer...");
        consumer.shutdown();
        log.info("DefaultMQPushConsumer destroy success.");
    }

    public DefaultMQPushConsumer getConsumer() {
        return consumer;
    }


}
